package com.eureka.provider.controller;

import com.alibaba.fastjson.JSON;
import com.eureka.provider.mapper.OrderMapper;
import com.eureka.provider.mapper.SeckillOrderMapper;
import com.eureka.provider.pojo.Order;
import com.eureka.provider.pojo.SeckillOrder;
import com.eureka.provider.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * <p>
 *  前端控制器
 * </p>
 *
 * @author sam
 * @since 2021-04-15
 */
@RestController
@Slf4j
@RequestMapping("/order")
public class OrderController {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @RequestMapping("/kafka")
    public void kafkaTest(String msg){
        kafkaTemplate.send("test",msg);
    }

    @KafkaListener(topics = {"test"})
    public void listener(ConsumerRecord<?,?> record){
        System.out.println("消费消息："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

    @Autowired
    OrderMapper orderMapper;
    @Autowired
    SeckillOrderMapper seckillOrderMapper;
    @Autowired
    IOrderService orderService;
    @Autowired
    RedisTemplate redisTemplate;

    @KafkaListener(topics = {"order"})
    public void orderConsumer(ConsumerRecord<?,?> record){
        Order order= JSON.parseObject(String.valueOf(record.value()),Order.class);
        int insert = orderMapper.insert(order);
        if(insert==1){
            redisTemplate.delete("goods:"+order.getGoodsId());
            log.info("订单生成成功："+order.toString());
        }
    }

    @KafkaListener(topics = {"seckillOrder"})
    public void seckillOrderConsumer(ConsumerRecord<?,?> record){
        SeckillOrder seckillOrder= JSON.parseObject(String.valueOf(record.value()),SeckillOrder.class);
        int insert = seckillOrderMapper.insert(seckillOrder);
        if(insert==1){
            log.info("秒杀订单生成成功："+seckillOrder.toString());
        }
    }
}
